package main

import (
	"crypto/md5"
	"flag"
	"fmt"
	"go.uber.org/zap"
	"gse/transport/udp/connection"
	"math/rand"
	"net"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
)

var messageSendRate int
var dataSendRate int
var totalMessageSent int64
var messageLengthRange [2]int
var messageSerialToken string
var totalMessageBytes int64
var requireAck bool
var logger *zap.Logger
var stopChan = make(chan bool)
var wg = new(sync.WaitGroup)

func main() {
	flagset := flag.NewFlagSet("parameters", flag.ExitOnError)
	var sAddr string
	flagset.StringVar(&sAddr, "s", "", "server address, e.g. 127.0.0.1:3131")

	flagset.BoolVar(&requireAck, "a", false, "whether requires the server to ack each data packet.")
	var keepAlive bool
	flagset.BoolVar(&keepAlive, "k", false, "keep alive.")

	flagset.IntVar(&dataSendRate, "n", 14000, "UDP packets sent per second")
	var retryInterval int
	flagset.IntVar(&retryInterval, "i", 1000, "interval of retries for failed sends.")

	var concurrency int
	flagset.IntVar(&concurrency, "c", 10, "number of clients to mock")

	flagset.IntVar(&messageSendRate, "mps", 10, "app message sent per second.")

	var messageLengthRangeStr string
	flagset.StringVar(&messageLengthRangeStr, "ml", "400-2048", "single message length in form of lowest-height")

	flagset.Parse(os.Args[1:])
	go http.ListenAndServe("0.0.0.0:6365", nil)
	addr, err := net.ResolveUDPAddr("udp4", sAddr)
	if err != nil {
		panic(err)
	}
	mlrp := strings.Split(messageLengthRangeStr, "-")
	if len(mlrp) != 2 {
		panic("invalid message length range")
	}
	rl, err := strconv.ParseInt(mlrp[0], 10, 64)
	if err != nil {
		panic(fmt.Errorf("invalid lower message length range: %v", err))
	}
	rh, err := strconv.ParseInt(mlrp[1], 10, 64)
	if err != nil {
		panic(fmt.Errorf("invalid higher message length range: %v", err))
	}
	messageLengthRange[0] = int(rl)
	messageLengthRange[1] = int(rh)
	if messageLengthRange[1] < messageLengthRange[0] {
		panic("invalid message length range")
	}

	messageSerialToken = fmt.Sprintf("%x", md5.Sum([]byte(time.Now().String())))

	fmt.Println("starting bench mark...")
	fmt.Println("sereial token: ", messageSerialToken)
	fmt.Println("concurrency：", concurrency)
	fmt.Println("message send rate：", messageSendRate)
	fmt.Println("message length range：", messageLengthRange)

	logger, _ = zap.NewDevelopment()
	st := time.Now()
	wg.Add(concurrency)
	for ; concurrency > 0; concurrency-- {
		c, err := connection.Connnect(addr)
		if err != nil {
			panic(err)
		}
		fmt.Printf("Connected to remote server[%v] with local address[%v]\n", c.RemoteAddr(), c.LocalAddr())
		cfg := c.GetConfig()
		cfg.PacketSendRate = dataSendRate
		cfg.SendRetryInterval = retryInterval
		cfg.AckTimeout = retryInterval / 3
		cfg.Logger = logger
		c.SetConfig(cfg)

		go getResponse(c)
		go getPing(c)
		go mockAppMessage(c)
		if keepAlive {
			c.KeepAlive()
		}

	}
	if !keepAlive {
		return
	}

	sigCh := make(chan os.Signal)
	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
	<-sigCh
	close(stopChan)
	wg.Wait()
	du := time.Now().Sub(st)
	logger.Debug(fmt.Sprintf("%v %v %v", "*********total message count：", atomic.LoadInt64(&totalMessageSent), "*****************"))
	logger.Debug(fmt.Sprintf("%v %v %v", "*********total message bytes：", atomic.LoadInt64(&totalMessageBytes), "*****************"))
	logger.Debug(fmt.Sprintf("%v %v %v", "*********message transfer rates：", float64(atomic.LoadInt64(&totalMessageBytes))/float64(du.Seconds()), "bytes/s *****************"))
	logger.Debug(fmt.Sprintf("%v %v %v", "*********time elapsed：", du, "*****************"))
}

func mockAppMessage(conn *connection.Connection) {
	var err error
	var message []byte
	defer wg.Done()
	for {
		select {
		case <-stopChan:
			return
		default:
		}
		rand.Seed(time.Now().UnixNano())
		message = []byte(messageSerialToken + ": " + fmt.Sprintf("%v", time.Now().UnixNano()) + " " + strings.Repeat("M", messageLengthRange[0]+rand.Intn(messageLengthRange[1]-messageLengthRange[0])))
		atomic.AddInt64(&totalMessageBytes, int64(len(message)))
		if requireAck {
			_, err = conn.SendDataWithAck([]byte(message))
		} else {
			_, err = conn.SendData([]byte(message))
		}

		if err != nil {
			fmt.Println(err)
		} else {
			logger.Debug(fmt.Sprintf("sent message: %v %v", conn.LocalAddr().String(), string(message)))
			atomic.AddInt64(&totalMessageSent, 1)
		}
		if messageSendRate > 0 {
			time.Sleep(time.Second / time.Duration(messageSendRate))
		}
	}
}

func getPing(conn *connection.Connection) {
	ticker := time.NewTicker(time.Second * 5)
	for {
		<-ticker.C

		duration, err := conn.Ping()
		prefix := "ping result: "
		data := []byte(fmt.Sprintf("%v, err: %v", duration, err))
		sep := strings.Repeat("=", len(prefix)+6)
		logger.Debug(fmt.Sprintf(`%v
|| %v %v ||
%v
`, sep, prefix, string(data), sep))
	}
}
func getResponse(conn *connection.Connection) {
	for {
		data, err := conn.ReadData()
		if err != nil {
			fmt.Printf("READ ERR: %v\n", err)
			continue
		}
		prefix := "server response: "
		sep := strings.Repeat("=", len(prefix)+6)
		logger.Debug(fmt.Sprintf(`%v
|| %v %v ||
%v
`, sep, prefix, string(data), sep))
	}
}
